package io.kiki.sba.registry.server.data_store.remoting.sessionserver.handler;


import io.kiki.sba.registry.api.Channel;
import io.kiki.sba.registry.common.model.ClientInterfaceId;
import io.kiki.sba.registry.common.model.PublishType;
import io.kiki.sba.registry.common.model.RegisterVersion;
import io.kiki.sba.registry.common.model.dataserver.BatchRequest;
import io.kiki.sba.registry.common.model.dataserver.ClientOffPublisher;
import io.kiki.sba.registry.common.model.dataserver.DatumVersion;
import io.kiki.sba.registry.common.model.slot.SlotAccess;
import io.kiki.sba.registry.common.model.slot.SlotAccessGenericResponse;
import io.kiki.sba.registry.common.model.store.Publisher;
import io.kiki.sba.registry.common.model.store.UnPublisher;
import io.kiki.sba.registry.server.data_store.change.DataChangeType;
import io.kiki.sba.registry.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executor;
import java.util.concurrent.ThreadPoolExecutor;

public class BatchPutDataHandler extends AbstractDataHandler<BatchRequest> {
    private static final Logger logger = LoggerFactory.getLogger(BatchPutDataHandler.class);
    @Autowired
    private ThreadPoolExecutor publishProcessorExecutor;

    @Override
    public void checkParam(BatchRequest request) {
        checkClientInterfaceId(request.getClientInterfaceId());
        for (Object object : request.getRequest()) {
            if (object instanceof Publisher) {
                checkPublisher((Publisher) object);
            } else
                if (object instanceof ClientOffPublisher) {
                    StringUtil.nullValue(((ClientOffPublisher) object).getClientToClientInterfaceServerNodeId(), "ClientOffPublisher.connectId");
                    StringUtil.nullValue(((ClientOffPublisher) object).getPublisherMap(), "ClientOffPublisher.publisherMap");
                } else {
                    throw new IllegalArgumentException("unsupported item in batch:" + object);
                }
        }
    }

    @Override
    public Object doHandle(Channel channel, BatchRequest request) {
        final ClientInterfaceId clientInterfaceId = request.getClientInterfaceId();
        processClientInterfaceId(channel, clientInterfaceId);

        return handleRequest(request, clientInterfaceId);
    }

    public SlotAccessGenericResponse<Object> handleRequest(BatchRequest batchRequest, ClientInterfaceId clientInterfaceId) {
        final SlotAccess slotAccess = checkSlotAccess(serverConfig.getDataCenter(), batchRequest.getSlotId(), batchRequest.getSlotsEpoch(), batchRequest.getSlotLeaderEpoch());
        if (slotAccess.isMoved() || slotAccess.isMisMatch()) {
            // only reject the when moved
            return SlotAccessGenericResponse.failedResponse(slotAccess);
        }
        final String slotIdStr = String.valueOf(batchRequest.getSlotId());
        final Set<String> changeDataInfoIds = new HashSet<>(128);
        try {
            for (Object req : batchRequest.getRequest()) {
                // contains publisher and unPublisher
                if (req instanceof Publisher) {
                    Publisher publisher = (Publisher) req;
                    DatumVersion updatedVersion = doHandle(publisher);
                    if (updatedVersion != null) {
                        changeDataInfoIds.add(publisher.getDataInfoId());
                    }
                    if (publisher instanceof UnPublisher) {
                        logger.info("unpub,{},{},{},{},{},{}", slotIdStr, publisher.getDataInfoId(), publisher.getRegisterId(), publisher.getVersion(), publisher.getRegisterTimestamp(), updatedVersion);
                    } else {
                        logger.info("pub,{},{},{},{},{},{}", slotIdStr, publisher.getDataInfoId(), publisher.getRegisterId(), publisher.getVersion(), publisher.getRegisterTimestamp(), updatedVersion);
                    }
                } else
                    if (req instanceof ClientOffPublisher) {
                        ClientOffPublisher clientOffPublisher = (ClientOffPublisher) req;
                        Map<String, DatumVersion> updatedVersion = doHandle(clientOffPublisher, clientInterfaceId);
                        changeDataInfoIds.addAll(updatedVersion.keySet());
                        for (Map.Entry<String, Map<String, RegisterVersion>> entry : clientOffPublisher.getPublisherMap().entrySet()) {
                            final String dataInfoId = entry.getKey();
                            for (Map.Entry<String, RegisterVersion> entry1 : entry.getValue().entrySet()) {
                                RegisterVersion version = entry1.getValue();
                                logger.info("off,{},{},{},{},{},{}", slotIdStr, dataInfoId, entry1.getKey(), version.getVersion(), version.getRegisterTimestamp(), updatedVersion.get(dataInfoId));
                            }
                        }
                    } else {
                        throw new IllegalArgumentException("unsupported item in batch:" + req);
                    }
            }
        } finally {
            // if has exception, try to notify the req which was handled
            if (!changeDataInfoIds.isEmpty()) {
                dataChangeEventCenter.onChange(changeDataInfoIds, DataChangeType.put, serverConfig.getDataCenter());
            }
        }

        return SlotAccessGenericResponse.successResponse(slotAccess, null);
    }

    private DatumVersion doHandle(Publisher publisher) {
        publisher = Publisher.internPublisher(publisher);
        if (publisher.getPublishType() == PublishType.temporary) {
            // create datum for the temp publisher, we need the datum.version for check ver
            datumStorageDelegate.createEmptyDatumIfAbsent(serverConfig.getDataCenter(), publisher.getDataInfoId());
            // temporary only notify session, not store
            dataChangeEventCenter.onTempPubChange(publisher, serverConfig.getDataCenter());
            return null;
        }
        return datumStorageDelegate.putPublisher(serverConfig.getDataCenter(), publisher);
    }

    public Map<String, DatumVersion> doHandle(ClientOffPublisher request, ClientInterfaceId clientInterfaceId) {
        Map<String, Map<String, RegisterVersion>> dataInfoIdTopublisherMap = request.getPublisherMap();
        Map<String, DatumVersion> map = new HashMap<>(dataInfoIdTopublisherMap.size());
        for (Map.Entry<String, Map<String, RegisterVersion>> entry : dataInfoIdTopublisherMap.entrySet()) {
            DatumVersion version = datumStorageDelegate.removePublishers(serverConfig.getDataCenter(), clientInterfaceId, entry.getKey(), entry.getValue());
            if (version != null) {
                map.put(entry.getKey(), version);
            }
        }
        return map;
    }

    @Override
    public Class interest() {
        return BatchRequest.class;
    }

    @Override
    public Executor getExecutor() {
        return publishProcessorExecutor;
    }
}
